package com.mq.agency.server;

import com.mq.agency.mq.MyConsumer;
import com.mq.agency.vo.Stu;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.commons.lang3.SerializationUtils;

import java.util.concurrent.ConcurrentHashMap;

public class ProxyServerHandler extends ChannelInboundHandlerAdapter {

    // 创建一个ChannelGroup，其是一个线程安全的集合，其中存放着与当前服务器相连接的所有Active状态的Channel
    // GlobalEventExecutor是一个单例、单线程的EventExecutor，是为了保证对当前group中的所有Channel的处理
    // 线程是同一个线程
    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    public static ConcurrentHashMap stationMap=new ConcurrentHashMap();

    // 只要有客户端Channel给当前的服务端发送了消息，那么就会触发该方法的执行
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Object deserialize = SerializationUtils.deserialize((byte[]) msg);
        System.out.println(deserialize);
        Stu stu = (Stu) deserialize;
        //根据获取的消息建立mq
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setHost(stu.getHost());
        connectionFactory.setPort(stu.getPort());
        connectionFactory.setUsername(stu.getUserName());
        connectionFactory.setPassword(stu.getPassword());
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();
        com.rabbitmq.client.Channel channel = connection.createChannel();
        /**
         * 声明一个队列
         * 第一个参数表示这个信道连接哪个队列
         * 第二个参数表示是否持久化，当这个参数设置为true，即使你的服务器关了从新开数据还是存在的
         * 第三个参数表示是否独占队列，也就是所只能自己去监听这个队列
         * 第四个参数表示队列脱离绑定时是否自动删除
         * 第五个参数表示扩展参数，可设置为null
         */
        channel.exchangeDeclare(stu.getExchangeName(),"topic",true,false,null);
        channel.queueDeclare(stu.getQueueName(),true,false,false,null);
        channel.queueBind(stu.getQueueName(),stu.getExchangeName(),stu.getRouteKey());
        //指该消费者在接收到队列里的消息但没有返回确认结果之前,
        // 队列不会将新的消息分发给该消费者。队列中没有被消费的消息不会被删除，还是存在于队列中。
        channel.basicQos(1);
        //创建消费者，指定建立在那个连接上
        MyConsumer myConsumer=new MyConsumer(channel);
        //设置channel
        // 第二个参数 是否自动签收  false的话要手动签收back
        // 第三个参数表示消费对象
        channel.basicConsume(stu.getQueueName(),true,myConsumer);
        //放入map
        stationMap.put(stu.getQueueName(),ctx.channel());
    }

    // 只要有客户端Channel与服务端连接成功就会执行这个方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 获取到当前与服务器连接成功的channel
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "---上线");
        group.writeAndFlush(channel.remoteAddress() + "---上线\n");
        // 将当前channel添加到group中
        group.add(channel);
    }

    // 只要有客户端Channel断开与服务端的连接就会执行这个方法
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 获取到当前要断开连接的Channel
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "------下线");
        group.writeAndFlush(channel.remoteAddress() + "下线，当前在线人数：" + group.size() + "\n");

        // group中存放的都是Active状态的Channel，一旦某Channel的状态不再是Active，
        // group会自动将其从集合中踢出，所以，下面的语句不用写
        // remove()方法的应用场景是，将一个Active状态的channel移出group时使用
        // group.remove(channel);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}
